package pipeline

import (
	"sort"
	"io"
	"encoding/binary"
	"math/rand"
)

//返回类型加上了<-表示数据只出不进
func ArraySource(a ...int) <-chan int {
	out:=make(chan int)
	go func(){
		for _,v:=range a {
			out<-v
		}
		close(out)
	}()
	return out
}
//参数加上<-表示管道数据有输入
func InMemSort(in <-chan int) <-chan int {
	out:=make(chan int)
	go func() {
		a:=[]int{}
		for v:=range in{
			a=append(a,v)
		}
		sort.Ints(a)
		for _,v:=range a{
			out<-v
		}
		close(out)
	}()
	return out
}
func Merge(in1,in2 <-chan int) <-chan int {
	out:=make(chan int)
	go func() {
		v1,ok1:=<-in1
		v2,ok2:=<-in2
		for ok1||ok2 {
			//如果in2空了，或者in1不为空并且v1<v2，将v1写入channel，继续读取in1并赋值给v1、ok1
			if !ok2 || (ok1&&v1<=v2) {
				out<-v1
				v1,ok1=<-in1
			}else {
				out<-v2
				v2,ok2=<-in2
			}
		}
		close(out)
	}()
	return out
}
//chunkSize表示文件分块后的每块大小
func ReaderSource(reader io.Reader,chunkSize int) <- chan int {
	out:=make(chan int)
	go func() {
		//读取八个字节，为一个uint64
		buffer:=make([]byte,8)

		bytesRead:=0
		for{
			n,err:=reader.Read(buffer)
			bytesRead+=n
			if n>0 {
				v:=int(binary.BigEndian.Uint64(buffer))
				out<-v
			}
			if err!=nil || (chunkSize!=-1 && bytesRead>=chunkSize){
				break
			}
		}
		close(out)
	}()
	return out
}
func WriteSink(writer io.Writer,in<-chan int ){
	for v:=range in {
		buffer:=make([]byte,8)
		binary.BigEndian.PutUint64(buffer,uint64(v))
		writer.Write(buffer)
	}
}
func RandomSource(count int) <-chan int {
	out:=make(chan int)
	go func() {
		for i:=0;i<count;i++ {
			out<-rand.Int()
		}
		close(out)
	}()
	return out
}
//归并多个chan
func MergeN(inputs ...<-chan int ) <-chan int {

	//如果只有一个chan
	if len(inputs)==1 {
		return inputs[0]
	}

	//多个chan分为两部分，并取得分界点
	m:=len(inputs)/2
	//递归归并
	out:=Merge(MergeN(inputs[:m]...),MergeN(inputs[m:]...))
	return out
}